{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "67e13836",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "551b0582",
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf().setAppName(\"demo\").setMaster(\"local[*]\")\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "89744bff",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 创建成RDD\n",
    "rdd1 = sc.parallelize([1,2,3,4,5])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "1f4d8513",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd2 = sc.textFile(\"file:///D:/text.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fb4e81e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#查看分区数据情况\n",
    "print(rdd1.glom().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "e7f457ee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2\n"
     ]
    }
   ],
   "source": [
    "#查看分区数\n",
    "print(rdd2.getNumPartitions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "61924baa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PythonRDD[9] at RDD at PythonRDD.scala:53\n"
     ]
    }
   ],
   "source": [
    "#word count\n",
    "result = rdd2.flatMap(lambda x: x.split(\" \")).map(lambda x: (x,1)).reduceByKey(lambda x,y: x+y)\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b876a9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 基本算子1 map 和 flatMap"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1e748e96",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 基本算子 glom"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "c88f0938",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(0, <pyspark.resultiterable.ResultIterable object at 0x000001BAA54586D0>), (1, <pyspark.resultiterable.ResultIterable object at 0x000001BAA8767FD0>)]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 groupBy\n",
    "rdd3 = rdd1.groupBy(lambda x: x%2)\n",
    "print(rdd3.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "a06eb545",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd3.foreach(lambda x: print(list(x[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b2ccfe21",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 基本算子 filter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "2efd9dc9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4, 5, 6, 7]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 distinct\n",
    "rdd4 = sc.parallelize([1,2,4,5,4,6,7,3,4,5,1,2,3,4,5])\n",
    "print(rdd4.distinct().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "7a3e20f4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 1, 2, 2, 3, 3, 4, 4, 4, 4, 5, 5, 5, 6, 7]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 sortBy\n",
    "rdd5 = rdd4.sortBy(lambda x: x, ascending=True)\n",
    "print(rdd5.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "c7051884",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4, 5, 1, 2, 4, 5, 4, 6, 7, 3, 4, 5, 1, 2, 3, 4, 5]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 union\n",
    "rdd6 = rdd1.union(rdd4)\n",
    "print(rdd6.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "264046c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4, 5]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 intersection\n",
    "rdd7 = rdd1.intersection(rdd4)\n",
    "print(rdd7.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "689380f1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', 1), ('b', 2), ('c', 3)]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 zip\n",
    "rdd8 = sc.parallelize(['a','b','c'])\n",
    "rdd9 = sc.parallelize([1,2,3])\n",
    "rdd10 = rdd8.zip(rdd9)\n",
    "print(rdd10.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "c9aa9055",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', 0), ('b', 1), ('c', 2)]\n",
      "[(1, 3), (2, 7), (3, 11)]\n"
     ]
    }
   ],
   "source": [
    "# 基本算子 zipWithIndex\n",
    "print(rdd8.zipWithIndex().collect())\n",
    "print(rdd9.zipWithUniqueId().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "855d2867",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 基本Action算子\n",
    "# collect, first, max, sum, count, take, top, foreach"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "b6db4ab1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[7, 6, 5]\n"
     ]
    }
   ],
   "source": [
    "# 基本Action算子 takeOrdered\n",
    "rdd = sc.parallelize([6,4,5,2,4,1,7])\n",
    "print(rdd.takeOrdered(3, lambda x: -x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "671684a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "56\n"
     ]
    }
   ],
   "source": [
    "# 聚合函数 reduce\n",
    "result = rdd4.reduce(lambda x,y: x+y)\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "3d4f335e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "85\n"
     ]
    }
   ],
   "source": [
    "# 聚合函数 fold \n",
    "rdd11 = sc.parallelize(range(10),3)\n",
    "result = rdd11.fold(10, lambda x,y: x+y)\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "2542d6c4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[83, 50, 96, 73, 92, 54, 36, 49, 46, 39]\n"
     ]
    }
   ],
   "source": [
    "# 抽样算子 takeSample\n",
    "rdd12 = sc.parallelize(range(100))\n",
    "rdd13 = rdd12.takeSample(False, 10, 1)\n",
    "print(rdd13)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "578bf735",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "27\n"
     ]
    }
   ],
   "source": [
    "# example1 \n",
    "# 输入一篇英文文章，统计文章中的数字字符出现次数\n",
    "rdd = sc.parallelize(['asdf2323reasd','df32fsd234','fgh3hDFd','34234234234234','fasdfasdbr45t4'])\n",
    "rdd1 = rdd.flatMap(lambda x: list(x))\n",
    "result = rdd1.filter(lambda x: x.isdigit()).count()\n",
    "print(result)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "b048508b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[34234234234234, 2323, 234]\n"
     ]
    }
   ],
   "source": [
    "# example 2\n",
    "# 找到英文中的所有数字，找到top3\n",
    "import re\n",
    "\n",
    "def findNumbers(s):\n",
    "    return re.findall(r'\\d+', s)\n",
    "\n",
    "result = rdd.flatMap(lambda x: findNumbers(x)).map(lambda x: int(x)).top(3)\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "5a54a8f2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[33, 145, 195]\n",
      "[9928, 9908, 9782]\n"
     ]
    }
   ],
   "source": [
    "# example 3\n",
    "# 从1w个数抽取100个数，并去除奇数最小的3个和偶数最大的3个\n",
    "rdd = sc.parallelize(range(10000))\n",
    "rdd1 = sc.parallelize(rdd.takeSample(False, 100))\n",
    "rdd2 = rdd1.filter(lambda x: x%2==1)\n",
    "print(rdd2.takeOrdered(3, lambda x: x))\n",
    "rdd3 = rdd1.filter(lambda x: x%2==0)\n",
    "print(rdd3.top(3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "1d104d1b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['hello', 'hello', 'hello', 'world', 'spark', 'python']\n"
     ]
    }
   ],
   "source": [
    "# example 4 \n",
    "# 读取文件，按照单词第二个字母排序\n",
    "rdd2 = sc.textFile(\"file:///D:/text.txt\")\n",
    "rdd3 = rdd2.flatMap(lambda x: x.split(\" \")).sortBy(lambda x: x[1])\n",
    "print(rdd3.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "acabdaf7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# example 5\n",
    "# 读取文件，找到子母钟包含r的单词，按照长短升序排序\n",
    "rdd4 = rdd2.flatMap(lambda x: x.split(\" \")).filter(lambda x: 'r' in x).sort"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
